package com.gjy.scheduling.monitor.processor;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.logging.log4j.LogManager;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 线程对象控制类
 */
public class ThreadLocalUtils {
    private static final AtomicInteger POOL_NUMBER = new AtomicInteger(0);
    public static final int MAX_THREAD = 10000;
    private static final CopyOnWriteArrayList<ExecutorService> markUsedService = new CopyOnWriteArrayList<>();

    private ThreadLocalUtils() {

    }


    /**
     * 文件分发线程池线程对象
     */
    private static ConcurrentLinkedQueue<ExecutorService> serviceThreadLocal = new ConcurrentLinkedQueue<>();

    /**
     * 返回ExecutorService线程变量,使用这方法之后必须调用offerExecutorService()
     * Executors.newCachedThreadPool();
     * Executors.newFixedThreadPool(1);
     *
     * @return ExecutorService
     */
    public static ExecutorService getExecutorService(String business, int number) {
        ExecutorService executorService = serviceThreadLocal.poll();
        if (executorService == null) {
            if (number > MAX_THREAD) {
                number = MAX_THREAD;
            }
            ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat(business + "-pool-"
                    + POOL_NUMBER.getAndIncrement() + "-thread-%04d").build();
            executorService = new ThreadPoolExecutor(0, number,
                    3, TimeUnit.HOURS, new SynchronousQueue<Runnable>(),
                    namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
        }
        markUsedService.add(executorService);
        return executorService;
    }

    public static ExecutorService getExecutorServiceBlocking(String business, int number) {
        ExecutorService executorService = serviceThreadLocal.poll();
        if (executorService == null) {
            if (number > MAX_THREAD) {
                number = MAX_THREAD;
            }
            ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat(business + "-pool-"
                    + POOL_NUMBER.getAndIncrement() + "-thread-%04d").build();
            executorService = new ThreadPoolExecutor(0, number,
                    3, TimeUnit.HOURS, new LinkedBlockingQueue<Runnable>(),
                    namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
        }
        markUsedService.add(executorService);
        return executorService;
    }

    /**
     * 把对象返回给线程对垒
     *
     * @param executorService
     * @return
     */
    public static void offerExecutorService(ExecutorService executorService) {
        if (executorService != null) {
            serviceThreadLocal.offer(executorService);
            markUsedService.remove(executorService);
        }
    }

    public static CopyOnWriteArrayList<ExecutorService> getMarkUsedService() {
        return markUsedService;
    }

    /**
     * 主线程停止
     */
    public static void stopPool() {
        for (ExecutorService executorService : serviceThreadLocal) {
            try {
                executorService.shutdown();
            } catch (Exception e) {
                LogManager.getLogger(ThreadLocalUtils.class).warn("ExecutorService 关闭线程队列异常");
            }
        }
        for (ExecutorService executorService : markUsedService) {
            try {
                markUsedService.remove(executorService);
            } catch (Exception e) {
                LogManager.getLogger(ThreadLocalUtils.class).warn("markUsedService 关闭线程队列异常");
            }
        }
    }
}
